package com.shujia.flink.window

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time

object Demo3EventTImeSessionWindow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    /**
     * java,1667444174000
     * java,1667444175000
     * java,1667444176000
     * java,1667444180000
     * java,1667444179000
     * java,1667444181000
     * java,1667444182000
     * java,1667444184000
     * java,1667444185000
     */

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //1、整理时间将时间字段取出来
    val tsDS: DataStream[(String, Long)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      val word: String = split(0)
      //时间字段
      val ts: Long = split(1).toLong
      (word, ts)
    })

    //2、告诉flink哪一个字段是时间字段
    val assDS: DataStream[(String, Long)] = tsDS.assignAscendingTimestamps(_._2)

    val kvDS: DataStream[(String, Int)] = assDS.map(kv => (kv._1, 1))


    /**
     * EventTimeSessionWindows: 事件时间会话窗口
     */
    kvDS
      .keyBy(_._1)
      .window(EventTimeSessionWindows.withGap(Time.seconds(5)))
      .sum(1)
      .print()


    env.execute()

  }

}
